package com.hqbzl.c1.code;

import lombok.extern.slf4j.Slf4j;

import java.sql.Time;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author hq
 */
@Slf4j
public class TestPool {
    public static void sleep(long t) {
        try {
            TimeUnit.SECONDS.sleep(t);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        final ThreadPool threadPool = new ThreadPool(5, 2, 1, TimeUnit.SECONDS, (q, t) -> {
            log.debug("wait");
//            q.put(t); // 无限等
//            q.offer(t, 1, TimeUnit.SECONDS);  // 限时等
//            log.debug("放弃任务: {}", t); //什么都不做，放弃
//            throw new IllegalStateException("throw exception"); //抛出异常
//            t.run(); // 用户线程自己执行
        });
        final long start = System.currentTimeMillis();
        final CountDownLatch latch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    sleep(60 * 1000);
                    log.debug("thread execute! {}", j);
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("System.currentTimeMillis() - start = " + (System.currentTimeMillis() - start));
        /*
            threadPool.add(() -> {
                System.out.println("in queue task 1");
            });
            threadPool.add(() -> {
                System.out.println("in queue task 2");
            });
            threadPool.execute(() -> {
                System.out.println("thread execute!");
            });
         */
    }

    public static void main2(String[] args) {
        final MyQueue<String> my = new MyQueue<String>(5);
        new Thread(() -> {
            final String pull = my.pull(3, TimeUnit.SECONDS);
            log.debug("pull is: {}", pull);
        }, "pull-thread").start();
        sleep(1);
        new Thread(() -> {
            log.debug("put value");
            my.put("val");
        }, "th-provider").start();

    }

    public static void main1(String[] args) {
        final MyQueue<String> my = new MyQueue<String>(5);
        new Thread(() -> {
            int index = 0;
            while (true) {
                String val = "val-" + index;
                log.debug("put: {}", val);
                my.put(val);
                sleep(1);
                index++;
            }
        }, "th-provider").start();
        /*   */
        new Thread(() -> {
            while (true) {
                final String take = my.take();
                sleep(2);
                log.debug("===================================================> consumer: {}", take);
            }
        }, "th-consumer").start();
    }

    interface RejectPolicy<T> {
        void reject(MyQueue<T> queue, T task);
    }

    static class ThreadPool {
        private MyQueue<Runnable> myQueue;
        private Set<Worker> workers = new HashSet<>();
        private int coreSize;
        private long timeout;
        private TimeUnit timeUnit;
        private int capacity;
        private RejectPolicy<Runnable> rejectPolicy;

        public void add(Runnable runnable) {
            myQueue.put(runnable);
        }

        public ThreadPool(int capacity, int coreSize, long timeout, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
            this.capacity = capacity;
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.timeUnit = timeUnit;
            this.rejectPolicy = rejectPolicy;
            myQueue = new MyQueue<>(capacity);
        }

        public void execute(Runnable task) {
            synchronized (workers) {
                if (workers.size() < coreSize) {
                    Worker worker = new Worker(task);
                    workers.add(worker);
                    //myQueue.put(worker);
                    worker.start();
                } else {
                    //myQueue.put(task);
                    myQueue.tryPut(rejectPolicy, task);
                    // 1 死待
                    // 2 等超时
                    // 3 放弃任务
                    // 4 抛出异常
                    // 5 用户线程自己执行
                }
            }
        }

        class Worker extends Thread {
            private Runnable task;

            private Worker(Runnable task) {
                this.task = task;
            }

            @Override
            public void run() {
                //while (task != null || (task = myQueue.take()) != null) {
                while (task != null || (task = myQueue.pull(timeout, timeUnit)) != null) {
                    try {
                        log.debug("run run run");
                        task.run();
                    } catch (Exception e) {
                        log.debug("worker run: {}", e.getMessage(), e);
                    } finally {
                        task = null;
                    }
                }
                // 这个worker 已经没事可以做了，可以结束自己了
                log.debug("worker end");
                synchronized (workers) {
                    workers.remove(this);
                }
            }
        }
    }


    static class MyQueue<T> {
        final ReentrantLock room = new ReentrantLock();
        final Condition provider = room.newCondition();
        final Condition consumer = room.newCondition();

        private Deque<T> queue;
        private int size;

        public MyQueue(int size) {
            this.size = size;
            queue = new ArrayDeque<>(size);
        }

        public T take() {
            room.lock();
            try {
                while (queue.isEmpty()) {
                    try {
                        log.debug("ququ empty");
                        consumer.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                final T elem = queue.remove();
                provider.signal();
                return elem;
            } finally {
                room.unlock();
            }
        }

        public T pullEx(long t, TimeUnit unit) {
            room.lock();
            try {
                long nanos = unit.toNanos(t);
                while (queue.isEmpty()) {
                    if (nanos <= 0) {
                        break;
                    }
                    try {
                        nanos = consumer.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                provider.signal();
                return null;
            } finally {
                room.unlock();
            }
        }

        public T pull(long t, TimeUnit unit) {
            final long lt = unit.toNanos(t);
            room.lock();
            try {
                final long start = System.nanoTime();
                long useTime = 0;
                while (queue.isEmpty()) {
                    long awaitTime = lt - useTime;
                    if (awaitTime <= 0) {
                        break;
                    }
                    try {
                        log.debug("pull time");
                        consumer.await(awaitTime, TimeUnit.NANOSECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    useTime = System.nanoTime() - start;
                }
                if (queue.isEmpty()) {
                    return null;
                } else {
                    final T e = queue.removeFirst();
                    provider.signal();
                    return e;
                }
            } finally {
                room.unlock();
            }
        }

        public void put(T t) {
            room.lock();
            try {
                while (queue.size() == size) {
                    try {
                        log.debug("ququ full queue.size(): {}, size: {}", queue.size(), size);
                        provider.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                queue.addLast(t);
                consumer.signal();
            } finally {
                room.unlock();
            }
        }

        public boolean offer(T t, long time, TimeUnit unit) {
            room.lock();
            try {
                long nanos = unit.toNanos(time);
                while (queue.size() == size) {
                    // 返回的是剩余时间
                    if (nanos <= 0) {
                        return false;
                    }
                    try {
                        nanos = provider.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                queue.addLast(t);
                consumer.signal();
                return true;
            } finally {
                room.unlock();
            }
        }

        public int size() {
            room.lock();
            try {
                return size;
            } finally {
                room.unlock();
            }
        }

        public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
            room.lock();
            try {
                if (queue.size() == size) {
                    log.debug("queue is full");
                    rejectPolicy.reject(this, task);
                } else {
                    queue.addLast(task);
                    consumer.signal();
                }
            } finally {
                room.unlock();
            }
        }
    }
}
